package com.silky.officeline.thread;


import com.silky.common.bean.DataSourceConfig;
import com.silky.common.bean.MultiLevelJoinConfig;
import com.silky.common.bean.NodeConfig;
import com.silky.common.constant.IConstant;
import com.silky.common.service.NodeService;
import com.silky.common.utils.PageList;
import com.silky.common.utils.ThreadSafeList;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 多级join线程
 */
@Slf4j
public class MultiLevelJoinThread extends AbstractThread{

    private final DataSourceConfig dataSourceConfig;
    private final MultiLevelJoinConfig multiLevelJoinConfig;
    private final List<Map<String, Object>> parentNodeDatas;
    private final PageList currentPageList;
    private final CountDownLatch currentCountDownLatch;
    private final int currentLevel;
    private final String parentNodeDataAlias;
    private NodeService nodeService ;
    private Map<String,List<Map<String,Object>>> dataSetMap ;//记录数据集上下文
    public MultiLevelJoinThread(DataSourceConfig dataSourceConfig,
                                MultiLevelJoinConfig multiLevelJoinConfig,
                                List<Map<String, Object>> parentNodeDatas,
                                PageList currentPageList, CountDownLatch currentCountDownLatch,
                                int currentLevel, String parentNodeDataAlias) {
        this.dataSourceConfig = dataSourceConfig;
        this.multiLevelJoinConfig = multiLevelJoinConfig;
        this.parentNodeDatas = parentNodeDatas;
        nodeService = new NodeService(dataSourceConfig);
        this.currentPageList =  currentPageList;
        this.currentCountDownLatch = currentCountDownLatch;
        this.currentLevel = currentLevel;
        this.parentNodeDataAlias = parentNodeDataAlias;
    }

    @Override
    public void run(){
        System.out.println("MultiLevelJoinThread当前线程："+Thread.currentThread().getName()+"开始执行");
        //如果父集合内为空，则记录错误返回
        if(parentNodeDatas==null ||parentNodeDatas.size() == 0){
            log.error("父集合为空，无法进行多级关联");
            System.out.println("MultiLevelJoinThread当前线程："+Thread.currentThread().getName()+"退出");
            currentCountDownLatch.countDown();
            return;
        }
        //计算当前join结果
        //构造LinkedBlockingQueue
        LinkedBlockingQueue<List<Map<String, Object>>> queue = new LinkedBlockingQueue<>();
        Thread[] threads = new Thread[multiLevelJoinConfig.getThreadNum()];
        int pageNum = currentPageList.size();
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(new JoinResultCalculationThread(dataSourceConfig,
                    multiLevelJoinConfig,currentPageList,queue,currentLevel,parentNodeDatas,parentNodeDataAlias));
            threads[i].start();
        }
        int joinCount = 0;
        System.out.println("MultiLevelJoinThread当前线程："+Thread.currentThread().getName()+"总页数:"+pageNum);
        while(joinCount<pageNum){
            try {
                //如果currentPageList为空，且queue为空，则退出
                if(currentPageList.size()==0 && queue.size()==0){
                    break;
                }
                List<Map<String,Object>> joinResult = queue.take();
                joinCount++;
                System.out.println("MultiLevelJoinThread当前线程："+Thread.currentThread().getName()+"已处理页数:"+joinCount+",queue.size:"+queue.size());
                //处理下一级
                if(currentLevel+1<=multiLevelJoinConfig.getNodes().size()-1){
                    NodeConfig nextNode = multiLevelJoinConfig.getNodes().get(currentLevel+1);
                    if(nextNode.getType().equals(IConstant.NODE_TYPE_JOIN)){
                        try {
                            processNextLevelJoin(joinResult,nextNode,currentLevel+1,multiLevelJoinConfig.getNodes().get(currentLevel).getJoinResultAlias());
                        } catch (Exception e) {
                            e.printStackTrace();
                            log.error("获取当前级别数据失败",e);
                            break;
                        }
                    }else if(nextNode.getType().equals(IConstant.NODE_TYPE_UPDATE)){
                        try {
                            processNextLevelUpdate(joinResult,nextNode,currentLevel+1,multiLevelJoinConfig.getNodes().get(currentLevel).getJoinResultAlias());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            log.error("更新下级数据失败",e);
                        }
                    }else if (nextNode.getType().equals(IConstant.NODE_TYPE_SINGLE)){
                        log.warn("没有意义，忽略！");
                    }

                }else{
                    log.warn("已经没有下级节点了，当前级别为{},节点数组长度为{}",currentLevel,multiLevelJoinConfig.getNodes().size());
                    break;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                log.error("多级关联线程被中断",e);
            }
        }
        //当前的CountDownlatch减1
        currentCountDownLatch.countDown();
        System.out.println("MultiLevelJoinThread当前线程："+Thread.currentThread().getName()+"退出");
    }



    /**
     * 执行下级节点为Update节点的情况
     * @param currentNodeDatas
     * @param nextNode
     * @param nextLevel
     * @param joinResultAlias
     */
    private void processNextLevelUpdate(List<Map<String, Object>> currentNodeDatas, NodeConfig nextNode, int nextLevel, String joinResultAlias) throws InterruptedException {
        ThreadSafeList threadSafeList = new ThreadSafeList();
        threadSafeList.init(currentNodeDatas);
        CountDownLatch currentCountDownLatch = new CountDownLatch(threadSafeList.size());
        for(int i=0;i<multiLevelJoinConfig.getThreadNum();i++) {
            MultiLevelUpdateThread multiLevelUpdateThread = new MultiLevelUpdateThread(dataSourceConfig, multiLevelJoinConfig, threadSafeList, nextNode, nextLevel,currentCountDownLatch);
            multiLevelUpdateThread.start();
        }
        //等待线程结束
        System.out.println("当前线程："+Thread.currentThread().getName()+"等待子线程processNextLevelUpdate执行完毕");
        currentCountDownLatch.await();
        System.out.println("当前线程："+Thread.currentThread().getName()+"子线程processNextLevelUpdate执行完毕");
    }

    /**
     *  处理下级节点为Join节点的情况
     * @param currentNodeDatas
     * @param nextNode
     * @param level
     * @param joinResultAlias
     */
    private void processNextLevelJoin(List<Map<String, Object>> currentNodeDatas, NodeConfig nextNode, int level, String joinResultAlias) throws Exception {
        //计算下一节点页数列表
        PageList nextPageList = new PageList();
        long totalCount = nodeService.getCount(nextNode);
        List<Long> offsets = nodeService.splitOffset(totalCount, nextNode.getBatchSize());
        nextPageList.init(offsets);
        //构造下级CountDownLatch
        CountDownLatch nextCountDownLatch = new CountDownLatch(nextPageList.size());
        Thread[] threads = new Thread[multiLevelJoinConfig.getThreadNum()];
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(new MultiLevelJoinThread(dataSourceConfig,multiLevelJoinConfig,currentNodeDatas,nextPageList,nextCountDownLatch,level, joinResultAlias));
            threads[i].start();
        }
        //等待子线程处理完毕
        nextCountDownLatch.await();
    }
}
